package com.foreveross.bsl.common.utils.balance.zookeeper;

import org.apache.commons.lang3.StringUtils;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.ZooKeeper.States;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

/**
 * ZK通用启动程序
 *
 * @author hyw
 */
public abstract class ZookeeperCommon {

    private static final Logger logger = LoggerFactory.getLogger(ZookeeperCommon.class);
    
    static int RECONNECT_SLEEP = 5000;
    String serverList = "localhost:2181";
    int sessionTimeout = 3000;
    String path;
    Set<String> paths;
    ZooKeeper zk;

    public ZookeeperCommon(String serverList, int sessionTimeout) {
        if (StringUtils.isBlank(serverList)) {
            throw new IllegalArgumentException("参数非法，serverList必填！");
        } if (sessionTimeout != 0) {
            this.sessionTimeout = sessionTimeout;
        }
        this.serverList = serverList;
        this.buildSimpleZK(serverList, sessionTimeout);
    }

    /** 生成一个简单的zk客户端
     * @param serverList
     * @param sessionTimeout
     */
    protected void buildSimpleZK(String serverList, int sessionTimeout) {
        try {
            final CountDownLatch connectedLatch = new CountDownLatch(1); // 同步器

            this.zk = new ZooKeeper(serverList, sessionTimeout, new Watcher() { // 启动监听
                @Override
                public void process(WatchedEvent event) {
                    logger.debug("ZKCommonKeeper WatchedEvent type={} state={}", event.getType(), event.getState());

                    if (event.getType() == Event.EventType.None) {
                        switch (event.getState()) { // 监听zk-client连接状态
                            case Expired: // 过期机制：重连并且绑定watch
                                // It's all over, reconnect...
                                logger.info("ZKSession过期，进行重新连接。。。");
                                boolean connected = false;
                                while (!connected) {
                                    try {
                                        rebuild(); // 重新构建zk客户端
                                        Thread.sleep(RECONNECT_SLEEP);
                                        if (States.CONNECTED == zk.getState()) {
                                            connected = true;
                                            bindWatcher(path);
                                            bindWatcher(paths);
                                            logger.info("ZKSession过期后重连成功，重新绑定Watcher，path={}", path);
                                        }
                                    } catch (InterruptedException e) {
                                        e.printStackTrace();
                                    }
                                }
                                break;
                            case SyncConnected: // 连接成功
                                // 避免KeeperErrorCode = ConnectionLoss /xxx
                                logger.info("ZKCommonKeeper Session SyncConnected.");
                                connectedLatch.countDown();
                                afterConnected();
                                break;
                            case AuthFailed:
                                logger.info("ZKCommonKeeper Session AuthFailed.");
                                break;
                            case ConnectedReadOnly:
                                logger.info("ZKCommonKeeper Session ConnectedReadOnly.");
                                break;
                            case Disconnected:
                                logger.info("ZKCommonKeeper Session Disconnected.");
                                break;
                            case SaslAuthenticated:
                                logger.info("ZKCommonKeeper Session SaslAuthenticated.");
                                break;
                            default:
                                break;
                        }
                    }
                }
            });
            // 等待zk服务线程完成连接
            waitUntilConnected(this.zk, connectedLatch);

            logger.info("ZooKeeper客户程序创建成功，serverlist={}，sessiontimeout={}",
                    serverList, sessionTimeout);

        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("无法构建zk实例：" + e.getMessage());
        }
    }

    /** 等待此线程操作完成
     * @param zooKeeper
     * @param connectedLatch
     */
    private void waitUntilConnected(ZooKeeper zooKeeper, CountDownLatch connectedLatch) {
        if (States.CONNECTING == zooKeeper.getState()) {
            try {
                logger.info("ZooKeeper客户程序正在等待连接，请稍后。。。");
                connectedLatch.await();
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            }
        }
    }

    /**
     *  重新连接zookeeper
     */
    public void rebuild() {
        try {
            if (zk != null) {
                zk.close();
            }
        } catch (InterruptedException e) {
           logger.error("关闭zk异常！", e);
        }
        this.buildSimpleZK(serverList, sessionTimeout);
    }

    abstract void bindWatcher(String path);

    abstract void bindWatcher(Set<String> paths);

    abstract void afterConnected();
}
